/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data625.handler.datadispatch;

import cn.ac.iie.ulss.data625.commons.RuntimeEnv;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.handler.AbstractHandler;

/**
 *
 * @author alexmu
 */
public class DataDispatchHandler extends AbstractHandler {

    //docs shcema
    private static Schema docsSchema = null;
    private static DatumReader<GenericRecord> docsReader = null;
    private static Map<String, String> filterIPs = null;
    private static boolean flowControl = false;
    private static int packageSizeRestraint = -1;
    private static String defaultRegion = null;

    private static DataDispatchHandler dataDispatchHandler = null;

    private Map<String, AtomicLong> dataSendStatistics = new HashMap<String, AtomicLong>();
    private static DataWriter dw = null;
    static Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = Logger.getLogger(DataDispatchHandler.class.getName());
    }

    private DataDispatchHandler() {
    }

    public static DataDispatchHandler getDataDispatchHandler() {
        if (dataDispatchHandler != null) {
            return dataDispatchHandler;
        }
        dataDispatchHandler = new DataDispatchHandler();
        try {
            Protocol protocol = Protocol.parse(new File("dx.json"));
            docsSchema = protocol.getType("docs");
            docsReader = new GenericDatumReader(docsSchema);
        } catch (IOException ex) {
            logger.error(ex.getMessage(), ex);
            dataDispatchHandler = null;
        }

        try {
            dw = new DataWriter();
        } catch (IOException ex) {
            logger.error(ex.getMessage(), ex);
            dataDispatchHandler = null;
        }

        try {
            String[] ipItems = RuntimeEnv.getParamAsString(RuntimeEnv.FILTER_IPS).split(",");
            if (ipItems.length > 0) {
                filterIPs = new HashMap<String, String>();
                for (String ipItem : ipItems) {
                    filterIPs.put(ipItem, ipItem);
                }
                logger.error("initing filter ip list is succeeded");
            } else {
                logger.info("no ip to filter");
            }
        } catch (Exception ex) {
            logger.error("initing filter ip list is failed for " + ex.getMessage(), ex);
            dataDispatchHandler = null;
        }

        flowControl = RuntimeEnv.getParamAsBoolean(RuntimeEnv.FLOW_CONTROL_ENABLE);
        packageSizeRestraint = RuntimeEnv.getParamAsInteger(RuntimeEnv.PACKAGE_SIZE_RESTRAINT);
        defaultRegion = RuntimeEnv.getParamAsString(RuntimeEnv.DEFAULT_REGION);

        if (flowControl) {
            TokenBucketManager.init();
        }

        SpeedMonitor.startMonitor();

        return dataDispatchHandler;
    }

    public static Map<String, Long> getDataVolumeStatics() throws Exception {
        Map<String, Long> dataVolumeStatisticSet = new HashMap<String, Long>();
        try {
            Iterator itor = dataDispatchHandler.dataSendStatistics.keySet().iterator();
            while (itor.hasNext()) {
                String docSchemaName = (String) itor.next();
                Long dataVolumeStatistc = dataDispatchHandler.dataSendStatistics.get(docSchemaName).longValue();
                dataVolumeStatisticSet.put(docSchemaName, dataVolumeStatistc);
            }
            return dataVolumeStatisticSet;
        } catch (Exception ex) {
            throw ex;
        }
    }

    @Override
    public void handle(String string, Request baseRequest, HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse) throws IOException, ServletException {
        long startTime = System.nanoTime();
        String reqID = String.valueOf(startTime);
        try {
            baseRequest.setHandled(true);

            String remoteHost = baseRequest.getRemoteAddr();
            int remotePort = baseRequest.getRemotePort();
            String region = getRegion(baseRequest);
            if (region.isEmpty()) {
                logger.warn("receive request from " + remoteHost + ":" + remotePort + " with no region set,so set default region " + defaultRegion);
                region = defaultRegion;
            }
            logger.info("receive request from " + remoteHost + ":" + remotePort + "@" + region + " and assigned id " + reqID);
            //是否过滤
            if (filterIPs.containsKey(remoteHost)) {
                String warnInfo = "req " + reqID + " is filtered for host " + remoteHost + " is not acceptable";
                logger.warn(warnInfo);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + warnInfo);
                return;
            }

            //retrive data
            ServletInputStream servletInputStream = null;
            byte[] req = null;
            try {
                logger.debug("req " + reqID + ":retriving bussisness data for request  ...");
                servletInputStream = baseRequest.getInputStream();
                ByteArrayOutputStream out = new ByteArrayOutputStream();
                byte[] b = new byte[4096];
                int i = 0;
                while ((i = servletInputStream.read(b, 0, 4096)) > 0) {
                    out.write(b, 0, i);
                }
                req = out.toByteArray();
                logger.debug("req " + reqID + ":length of bussisness data is " + req.length);
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":retrive bussiness data unsuccessfully for " + ex.getMessage();
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
                return;
            } finally {
                try {
                    servletInputStream.close();
                } catch (Exception ex) {
                }
            }
            //接到的数据是空的
            if (req == null || req.length == 0) {
                String warnInfo = "req " + reqID + ":retrive bussiness data unsuccessfully for content is empty";
                logger.error(warnInfo);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + warnInfo);
                return;
            }
            //接到的数据过长
            if (req.length > packageSizeRestraint) {
                String warnInfo = "req " + reqID + ":retrive bussiness data unsuccessfully for content size is more than " + packageSizeRestraint;
                logger.error(warnInfo);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + warnInfo);
                return;
            }
            //decode data
            ByteArrayInputStream docsbis = null;
            GenericRecord docsRecord = null;
            try {
                logger.debug("req " + reqID + ":decoding bussisness data ...");
                docsbis = new ByteArrayInputStream(req);//tuning
                BinaryDecoder docsbd = new DecoderFactory().binaryDecoder(docsbis, null);
                docsRecord = new GenericData.Record(docsSchema);
                docsReader.read(docsRecord, docsbd);
                logger.debug("req " + reqID + ":decode bussisness data sccessfully");
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":decode bussiness data unsuccessfully for " + ex.getMessage();
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
                return;
            } finally {
                try {
                    docsbis = null;
                } catch (Exception ex) {
                }
            }

            //check format of docs         
            String docSchemaName = null;
            try {
                docSchemaName = docsRecord.get("doc_schema_name").toString();
                if (docSchemaName == null || docSchemaName.isEmpty()) {
                    String errInfo = "req " + reqID + ":docSchemaName is empty";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                    return;
                } else if (!docSchemaName.equalsIgnoreCase("dx")) {
                    String errInfo = "req " + reqID + ":docSchemaName is " + docSchemaName + " and what is it???";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                    return;
                }
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":wrong format of bussiness data,can't get docSchemaName";
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
                return;
            }

            String docSchemaFullName = docSchemaName + "@" + region;
            logger.debug("req " + reqID + ":doc shcema full name is " + docSchemaFullName);

            try {
                //stat receive speed
                SpeedMonitor.incSchemaRecvVol(docSchemaFullName, req.length);

                String bzSysSign = docsRecord.get("sign").toString();

                GenericArray docsSet = (GenericData.Array<GenericRecord>) docsRecord.get("doc_set");
                if (docsSet == null) {
                    String errInfo = "req " + reqID + ":wrong format of bussiness data, doc_set is null";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                } else if (docsSet.size() <= 0) {
                    String errInfo = "req " + reqID + ":doc_set is empty";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                } else {
                    //flow control
                    if (flowControl) {
                        if (!TokenBucketManager.consumeTokenIfEnough(docSchemaFullName, req.length)) {
                            String info = "Flow control errInfo: schema " + docSchemaFullName + " have doesn't have enough token to be sent.";
                            logger.warn(info);
                            httpServletResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                            httpServletResponse.getWriter().println("-1\n" + "overload");
                            return;
                        }
                    }
                    try {
                        //写文件的地方
                        dw.writeData(docsSet);
                        //stat send speed
                        SpeedMonitor.incSchemaSendVol(docSchemaFullName, req.length);
                        logger.info("req " + reqID + ":writting " + docsSet.size() + " records of " + docSchemaFullName + ":" + bzSysSign + ":" + req.length + " to file successfully");
                        httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                        httpServletResponse.getWriter().println("0\n" + bzSysSign + "\n" + "OK");
                    } catch (Exception ex) {
                        String errInfo = "req " + reqID + ":internal error for sending data unsuccessfully for " + ex.getMessage();
                        logger.error(errInfo, ex);
                        httpServletResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                        httpServletResponse.getWriter().println("-1\n" + errInfo);
                    }
                }
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":internal error for sending data unsuccessfully for " + ex.getMessage();
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
            }
        } finally {
            long endTime = System.nanoTime();
            logger.info("dispose req " + reqID + " consume " + (endTime - startTime) / (1000 * 1000) + " ms");
        }
    }

    private byte[] toBytes(GenericArray pDocSet, Schema pSchema) {
        byte[] bytes = null;
        try {
            DatumWriter<GenericArray> docsWriter = new GenericDatumWriter<GenericArray>(pSchema);
            ByteArrayOutputStream docssos = new ByteArrayOutputStream();
            BinaryEncoder docsbe = new EncoderFactory().binaryEncoder(docssos, null);
            docsWriter.write((GenericArray) pDocSet, docsbe);
            docsbe.flush();
            return docssos.toByteArray();
        } catch (Exception ex) {
            ex.printStackTrace();
            bytes = null;
        }
        return bytes;
    }

    private String getRegion(Request req) {
        Object val = req.getParameter("region");
        return val == null ? "" : ((String) val).trim().toLowerCase();
    }

    public void shutdown() {
    }

}
